package com.taotao.job.executor.thread;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.taotao.job.core.localstore.LocalStore;
import com.taotao.job.core.localstore.RecoveryMessage;
import lombok.extern.slf4j.Slf4j;
import org.rocksdb.RocksDBException;

import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.lang.reflect.ParameterizedType;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 异步重试处理线程
 *
 * @param <T> 回调数据类型
 */
@Slf4j
public abstract class AbstractAsyncProccessThread<T extends Serializable> {

    protected static final ConcurrentHashMap<Class<? extends AbstractAsyncProccessThread>, AbstractAsyncProccessThread> instanceMap = new ConcurrentHashMap<>();

    public static AbstractAsyncProccessThread getInstance(Class<? extends AbstractAsyncProccessThread> clazz) {
        return instanceMap.get(clazz);
    }


    /**
     * 数据处理
     *
     * @param <T>
     */
    public interface Proccessor<T> {
        /**
         * 处理逻辑
         *
         * @param datas
         * @return
         */
        boolean proccess(List<T> datas) throws Exception;
    }

    /**
     * 失败数据本地存储
     */
    protected LocalStore recoveryStore;


    protected String name;

    protected Proccessor<T> proccessor;


    protected AbstractAsyncProccessThread(String name, Proccessor<T> proccessor) {
        this.name = name;
        this.proccessor = proccessor;

        this.asyncProccessThreadName = name + "-asyncProcessThread";
        this.asyncRetryThreadName = name + "-asyncRetryThread";
    }

    /**
     *
     * thread
     */
    private String asyncProccessThreadName;
    private String asyncRetryThreadName;
    private Thread asyncProcessThread;
    private Thread asyncRetryThread;
    private volatile boolean toStop = false;
    protected LinkedBlockingQueue<T> dataQueue = new LinkedBlockingQueue<>();

    public void pushData(T data) {
        dataQueue.add(data);
    }


    public void start() {

        this.recoveryStore = new LocalStore(name);


        // callback
        asyncProcessThread = new Thread(new Runnable() {

            @Override
            public void run() {

                // normal callback
                while (!toStop) {
                    try {
                        T callData = dataQueue.take();
                        if (callData != null) {

                            // callData list param
                            List<T> callbackParamList = new ArrayList<>();
                            int drainToNum = dataQueue.drainTo(callbackParamList);
                            callbackParamList.add(callData);

                            // callData, will retry if error
                            if (callbackParamList != null && callbackParamList.size() > 0) {
                                doProccess(callbackParamList);
                            }
                        }
                    } catch (Exception e) {
                        if (!toStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }

                // last callback
                try {
                    List<T> callbackParamList = new ArrayList<>();
                    int drainToNum = dataQueue.drainTo(callbackParamList);
                    if (callbackParamList != null && callbackParamList.size() > 0) {
                        doProccess(callbackParamList);
                    }
                } catch (Exception e) {
                    if (!toStop) {
                        log.error(e.getMessage(), e);
                    }
                }
                log.info(">>>>>>>>>>> xxl-job, executor callback thread destory.");

            }
        });
        asyncProcessThread.setDaemon(true);
        asyncProcessThread.setName(asyncProccessThreadName);
        asyncProcessThread.start();


        // retry
        asyncRetryThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!toStop) {
                    try {
                        retryRecoveryMessage();
                    } catch (Exception e) {
                        if (!toStop) {
                            log.error(e.getMessage(), e);
                        }

                    }
                    try {
                        TimeUnit.SECONDS.sleep(5000);
                    } catch (InterruptedException e) {
                        if (!toStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
                log.info(">>>>>>>>>>> xxl-job, executor retry logcallback thread destory.");
            }
        });
        asyncRetryThread.setDaemon(true);
        asyncRetryThread.setName(asyncRetryThreadName);
        asyncRetryThread.start();

    }

    public void toStop() {
        toStop = true;
        // stop callback, interrupt and wait
        if (asyncProcessThread != null) {    // support empty admin address
            asyncProcessThread.interrupt();
            try {
                asyncProcessThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }

        // stop retry, interrupt and wait
        if (asyncRetryThread != null) {
            asyncRetryThread.interrupt();
            try {
                asyncRetryThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }

    }

    /**
     * @param datas
     */
    private void doProccess(List<T> datas) {
        if (datas == null || datas.size() == 0) {
            return;
        }

        String key = Objects.hash(datas) + "";
        RecoveryMessage recoveryMessage = new RecoveryMessage(key, 0, JSON.toJSONString(datas), 1);

        doProccess(recoveryMessage);
    }

    /**
     * do callback, will retry if error
     *
     * @param recoveryMessage
     */
    private void doProccess(RecoveryMessage recoveryMessage) {
        boolean callbackRet = false;
        try {

            List<T> dataList = JSONObject.parseObject(recoveryMessage.getJson(), new TypeReference<List<T>>() {
            });

            if (proccessor.proccess(dataList)) {
                callbackRet = true;
            }
        } catch (Exception e) {
            log.error("", e);
        }


        if (!callbackRet) {
            recoveryMessage.incrementRecoveryCount();
            try {
                recoveryStore.put(recoveryMessage.getKey(), recoveryMessage);
            } catch (Exception e) {
                log.error("", e);
            }
        }
    }


    /**
     * 重试逻辑
     */
    private void retryRecoveryMessage() throws Exception {

        for (; ; ) {
            List<RecoveryMessage> recoveryMessageList = recoveryStore.queryList(30,RecoveryMessage.class);
            if (recoveryMessageList == null || recoveryMessageList.size() == 0) {
                return;
            }

            for (RecoveryMessage msg : recoveryMessageList) {
                doProccess(msg);
            }

        }

    }


    /**
     * 获取泛型T的类型
     *
     * @return
     */
    protected Class<T> getTClass() {
        Class<T> tClass = (Class<T>) ((ParameterizedType) getClass().getGenericSuperclass()).getActualTypeArguments()[0];
        return tClass;
    }

}
